package org.pupil.rpc.provider;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.pupil.rpc.annotation.RpcService;
import org.pupil.rpc.common.RpcServiceNameBuilder;
import org.pupil.rpc.common.ServiceMeta;
import org.pupil.rpc.config.RpcProperties;
import org.pupil.rpc.config.RpcServiceConfig;
import org.pupil.rpc.filter.FilterConfig;
import org.pupil.rpc.protocol.compress.Compress;
import org.pupil.rpc.protocol.compress.CompressFactory;
import org.pupil.rpc.protocol.handler.service.RpcRequestHandler;
import org.pupil.rpc.protocol.handler.service.ServiceAfterFilterHandler;
import org.pupil.rpc.protocol.handler.service.ServiceBeforeFilterHandler;
import org.pupil.rpc.protocol.codec.RpcDecoder;
import org.pupil.rpc.protocol.codec.RpcEncoder;
import org.pupil.rpc.protocol.serialization.SerializationFactory;
import org.pupil.rpc.registry.RegistryFactory;
import org.pupil.rpc.registry.RegistryService;
import org.pupil.rpc.router.LoadBalancerFactory;
import org.pupil.rpc.tolerant.FaultTolerantFactory;
import org.pupil.rpc.utils.PropertiesUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;


/**
 * 服务提供方后置处理器
 */
public class ProviderPostProcessor implements InitializingBean, BeanPostProcessor, EnvironmentAware {

    private Logger logger = LoggerFactory.getLogger(ProviderPostProcessor.class);

    // 配置类
    RpcProperties rpcProperties;

    // 此处在linux环境下改为0.0.0.0
    // 因为本地测试的原因用127.0.0.1  正常要通过配置类配置
    // private static String serverAddress = "127.0.0.1";


    /**
     * bean初始化后完成执行的逻辑
     *
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        // 创建线程开启RpcServer()
        Thread t = new Thread(() -> {
            try {
                startRpcServer();
            } catch (Exception e) {
                logger.error("start rpc server error.", e);
            }
        });
        t.setDaemon(true); // 设置为守护线程
        t.start();  // 开启线程
        // 初始化序列化
        SerializationFactory.init();
        //初始化压缩
        CompressFactory.init();
        // 初始化注册中心
        RegistryFactory.init();
        // 初始化负载均衡
        LoadBalancerFactory.init();
        // 初始化服务端的拦截器链
        FilterConfig.initServiceFilter();
        // 初始化容错策略
        FaultTolerantFactory.init();
    }

    /**
     * 启动服务方法
     *
     * @throws InterruptedException 处理线程中断异常
     */
    private void startRpcServer() throws InterruptedException {
        // 获取netty端口
        int serverPort = rpcProperties.getPort();

        // 获取nett
        String serverAddress = rpcProperties.getAddress();

        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup worker = new NioEventLoopGroup();

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(boss, worker)
                    .option(ChannelOption.SO_KEEPALIVE, true) //开启服务端ServerSocketChannel的心跳保活机制
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<NioSocketChannel>() {
                        @Override
                        protected void initChannel(NioSocketChannel channel) throws Exception {
                            channel.pipeline()
                                    .addLast(new RpcEncoder())
                                    .addLast(new RpcDecoder())
                                    .addLast(new ServiceBeforeFilterHandler()) // 服务前置拦截器
                                    .addLast(new RpcRequestHandler())    // 请求处理器
                                    .addLast(new ServiceAfterFilterHandler()); // 服务后置拦截器
                        }
                    })
                    .childOption(ChannelOption.SO_KEEPALIVE, true); //开启服务端SocketChannel的心跳保活机制

            ChannelFuture channelFuture = serverBootstrap.bind(serverAddress, serverPort).sync();
            logger.info("server addr {} started on port {}", serverAddress, serverPort);
            // 线程阻塞，等待关闭
            channelFuture.channel().closeFuture().sync();
            // 向 Java 虚拟机注册一个关闭钩子（shutdown hook）线程，在 JVM 即将关闭时会被自动启动并运行，用于执行一些清理或者释放资源的操作。
            Runtime.getRuntime().addShutdownHook(new Thread(() ->
            {
                logger.info("ShutdownHook execute start...");
                logger.info("Netty NioEventLoopGroup shutdownGracefully...");
                logger.info("Netty NioEventLoopGroup shutdownGracefully2...");
                // 安全释放EventLoopGroup
                boss.shutdownGracefully();
                worker.shutdownGracefully();
                logger.info("ShutdownHook execute end...");
            }, "Allen-thread"));
        } finally {
            // 安全释放EventLoopGroup
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }

    }


    /**
     * 服务注册
     * bean初始化前的后置处理器
     *
     * @param bean
     * @param beanName
     * @return
     * @throws BeansException
     */
    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        // 获取bean的class
        Class<?> beanClass = bean.getClass();
        // 找到bean上带有 RpcService 注解的
        RpcService rpcService = beanClass.getAnnotation(RpcService.class);
        if (rpcService != null) {
            // 获取bean实现的第一个接口名
            String serviceName = beanClass.getInterfaces()[0].getName();
            if (!rpcService.serviceInterface().equals(void.class)) {
                // 获得RpcService注解中配置的接口名
                serviceName = rpcService.serviceInterface().getName();
            }

            // 获取服务版本
            String serviceVersion = rpcService.serviceVersion();

            // 服务注册
            try {
                // 获取服务端口号
                int servicePort = rpcProperties.getPort();
                // 获取服务地址
                String serviceAddress = rpcProperties.getAddress();

                // 获取注册中心 ioc
                RegistryService registryService = RegistryFactory.get(rpcProperties.getRegisterType());
                // 创建服务信息
                ServiceMeta serviceMeta = new ServiceMeta();
                serviceMeta.setServiceAddr(serviceAddress);
                serviceMeta.setServicePort(servicePort);
                serviceMeta.setServiceName(serviceName);
                serviceMeta.setServiceVersion(serviceVersion);
                registryService.register(serviceMeta);
                // 缓存 注册成功的服务
                RpcServiceConfig.rpcServiceMap.put(RpcServiceNameBuilder.buildServiceKey(serviceMeta.getServiceName(), serviceMeta.getServiceVersion()), bean);
                logger.info("register server {} version {}", serviceName, serviceVersion);
            } catch (Exception e) {
                logger.error("failed to register service {}", serviceVersion, e);
            }
        }
        return bean;
    }


    /**
     * application启动时读取 属性配置文件的数据
     *
     * @param environment
     */
    @Override
    public void setEnvironment(Environment environment) {
        RpcProperties properties = RpcProperties.getInstance();
        PropertiesUtils.init(properties,environment);
        rpcProperties = properties;
        logger.info("读取配置文件成功");
    }
}
